/**
 * Copyright (C) 2010 Neofonie GmbH
 *
 * This programm is free software; you can redistribute it and/or modify
 * it under the terms of the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package eu.dicodeproject.analysis.histogram;

import java.io.IOException;
import java.text.FieldPosition;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Locale;

import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.Mapper;

/**
 * Scans over some HBase table. Looks for pre-configured queries in each
 * textCol. Queries are assigned integers, those integers are used as first part
 * of the output keys. Second part of the output keys correspond to bucket ids -
 * currently a new bucket is opened every week. Output values are equal to one
 * if a match was found in the tweet text for the respective query/bucket id.
 */
public class HBaseHistogramMapper extends TableMapper<IntWritable, IntWritable> {
  /** Set of patterns with associated key for later output. */
  public static enum ErrorCases {
    TWEETNULL, DATENULL
  };

  /** The year we started tracking twitter - our year zero. */
  private static final int TRACKING_START_YEAR = 2010;
  /** "Average" amount of days in a year. Ignoring leap years etc for now. */
  private static final int DAYS_IN_YEAR = 365;

  /** Date formatters for x creation. */
  private SimpleDateFormat dateFormat;
  private SimpleDateFormat dayOfYear;
  private SimpleDateFormat year;

  /** IntWritables for reuse */
  private IntWritable one = new IntWritable(1);
  private IntWritable resultKey = new IntWritable(0);

  @Override
  protected void setup(Context context) throws IOException, InterruptedException {
    super.setup(context);
    // Wed, 15 Dec 2010 09:39:23 +0000
    this.dateFormat = new SimpleDateFormat("EEE, dd MMM yyyy HH:mm:ss Z", Locale.ENGLISH);
    this.dayOfYear = new SimpleDateFormat("DDD", Locale.ENGLISH);
    this.year = new SimpleDateFormat("yyyy", Locale.ENGLISH);
  }

  @Override
  protected void map(final ImmutableBytesWritable row, final Result values, final Mapper.Context context)
      throws IOException, InterruptedException {
    // extract the data
    byte[] creationDateBytes = values.getValue(Bytes.toBytes("metaFamily"), Bytes.toBytes("creationDate"));
    byte[] tweetTextBytes = values.getValue(Bytes.toBytes("textFamily"), Bytes.toBytes("text"));

    // we anticipate errors as in non-existant tweetText or creationDate - those
    // are ignored
    boolean failed = false;
    if (tweetTextBytes == null) {
      context.getCounter(ErrorCases.TWEETNULL).increment(1);
      failed = true;
    }
    if (creationDateBytes == null) {
      context.getCounter(ErrorCases.DATENULL).increment(1);
      failed = true;
    }

    if (!failed) {

      String creationDate = new String(creationDateBytes);
      System.out.println(creationDate);
      Date date = null;
      try {
	date = this.dateFormat.parse(creationDate);
      } catch (ParseException e) {
	throw new IOException(e);
      }

      StringBuffer dayOfYearBuf = new StringBuffer();
      this.dayOfYear.format(date, dayOfYearBuf, new FieldPosition(0));
      int day = Integer.parseInt(dayOfYearBuf.toString());

      StringBuffer yearBuf = new StringBuffer();
      this.year.format(date, yearBuf, new FieldPosition(0));
      int year = Integer.parseInt(yearBuf.toString()) - TRACKING_START_YEAR;

      int x = year * DAYS_IN_YEAR + day;
      resultKey.set(x);
      context.write(this.resultKey, this.one);
    }
  }

}
